package com.archive.core;


import com.archive.core.config.ArchiveProperties;
import com.archive.core.constants.ArchiveStrategy;
import com.archive.core.exception.ArchiveException;
import com.archive.core.executor.OriginConnectionProvider;
import com.archive.core.processor.ArchiveProcessor;
import com.archive.core.table.ArchiveColumn;
import com.archive.core.table.ArchiveLock;
import com.archive.core.table.ArchiveTable;
import com.archive.core.util.ArchiveUtil;
import com.archive.core.util.Assert;
import com.archive.core.util.SystemClock;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.lang.reflect.Field;
import java.sql.*;
import java.util.*;
import java.util.regex.Matcher;

@Data
@Slf4j
public class ArchiveContext {

    private ArchiveProperties archiveProperties;
    private DataSource dataSource;
    private DataSource targetDataSource;
    private Map<String, ArchiveTable> archiveTableMap;
    private Field[] archiveTableFields;
    private static final String COLUMN_KEY_PRI = "PRI";

    private OriginConnectionProvider originConnectionProvider;

    public ArchiveContext() {
    }

    public ArchiveContext(ArchiveProperties archiveProperties) {
        this.archiveProperties = archiveProperties;
        this.archiveTableMap = new HashMap<>();
    }

    public void init() throws SQLException, IllegalAccessException {
        init(this.archiveProperties, this.archiveTableMap);
    }

    private boolean init(ArchiveProperties archiveProperties, Map<String, ArchiveTable> archiveTableMap) throws SQLException, IllegalAccessException {
        Integer archiveInterval = archiveProperties.getArchiveInterval();
        Assert.notNull(archiveInterval, () -> "archiveInterval can not be null.");
        if (archiveInterval <= 0) {
            throw new ArchiveException("archiveInterval must be greater than 0");
        }
        Map<String, ArchiveTable> propertiesArchiveTables = new HashMap<>();
        extractArchiveTable(archiveProperties.getArchiveTables(), propertiesArchiveTables);
        if (propertiesArchiveTables.isEmpty()) {
            log.info("No archive table found, archive disabled.");
            archiveProperties.setEnabled(false);
            return false;
        }

        initTableInfo(propertiesArchiveTables.keySet(), archiveTableMap);
        checkTableInfo(propertiesArchiveTables, archiveTableMap, archiveProperties.isCheckTable());
        if (archiveTableMap.isEmpty()) {
            log.info("No archive table found, archive disabled.");
            archiveProperties.setEnabled(false);
            return false;
        }

        archiveTableFields = ArchiveTable.class.getDeclaredFields();
        Arrays.stream(archiveTableFields).forEach(f -> f.setAccessible(true));

        // 将配置中的表依赖关系转到数据库表
        Map<String, ArchiveTable> archiveTableWithLevels = new HashMap<>();
        mergeArchiveTableConfig(archiveProperties.getArchiveTables(), null, archiveTableMap, archiveTableWithLevels,
                archiveProperties.isUseInsertIntoSelect(), archiveProperties.getArchiveStrategy());
        archiveTableMap.clear();
        archiveTableMap.putAll(archiveTableWithLevels);
        return true;
    }

    public void startArchive() {
        startArchive(this.archiveProperties, this.archiveTableMap);
    }

    public void startArchive(ArchiveProperties archiveProperties) throws ArchiveException {
        Map<String, ArchiveTable> archiveTableMap = new HashMap<>();
        try {
            if (!init(archiveProperties, archiveTableMap)) {
                return;
            }
        } catch (Exception e) {
            throw new ArchiveException(e);
        }
        startArchive(archiveProperties, archiveTableMap);
    }

    private void startArchive(ArchiveProperties archiveProperties, Map<String, ArchiveTable> archiveTableMap) throws ArchiveException {
        if (!archiveProperties.isEnabled()) {
            log.info("archive disabled.");
            return;
        }

        boolean locked = true;
        try {
            Connection connection = dataSource.getConnection();
            try {
                // 加锁 + 自动解锁
                String lockTableName = archiveProperties.getLockTableName();
                ArchiveLock archiveLock = lock(connection, lockTableName);
                if (archiveLock != null) {
                    Long lockTime = archiveLock.getLockTime();
                    Integer lockDeleteInterval = archiveProperties.getLockReleaseInterval();
                    // 自动解锁
                    if (lockDeleteInterval != null && lockDeleteInterval > 0
                            && SystemClock.now() - lockTime > (lockDeleteInterval * 24 * 60 * 60 * 1000L)) {
                        Long id = archiveLock.getId();
                        try (Statement statement = connection.createStatement()) {
                            statement.execute("delete from " + lockTableName + " where id = " + id + " and lock_time  = " + lockTime);
                            //
                            locked = statement.getUpdateCount() > 0;
                            log.info("auto release lock, lock id: {}, lock time: {}, release result: {}", id, lockTime, locked);
                        }
                    } else {
                        locked = false;
                    }
                }

                // 未被锁定或自动解锁成功
                if (locked) {
                    StringBuilder sql = new StringBuilder();
                    sql.append("insert into ").append(lockTableName).append("(id, lock_time) values(1, ").append(SystemClock.now()).append(")");
                    try (Statement stmt = connection.createStatement()) {
                        stmt.execute(sql.toString());
                    } catch (SQLIntegrityConstraintViolationException e) {
                        // 并发失败，退出
                        locked = false;
                    }
                }

                if (!locked) {
                    log.info("Another process is running.");
                    return;
                }
            } finally {
                ArchiveUtil.closeConnection(connection);
            }

            process(archiveProperties, ArchiveUtil.getArchiveTimeStamp(archiveProperties.getArchiveInterval(),
                    archiveProperties.getArchiveIntervalType()), archiveTableMap);
        } catch (Exception e) {
            throw new ArchiveException(e);
        } finally {
            try {
                // 解锁
                if (locked) {
                    try (Connection connection = dataSource.getConnection();
                         Statement stmt = connection.createStatement()) {
                        stmt.execute("delete from ks_archive_lock");
                    }
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    private static ArchiveLock lock(Connection connection, String lockTableName) throws SQLException {
        StringBuilder sql = new StringBuilder("select TABLE_SCHEMA,TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = '");
        sql.append(connection.getCatalog()).append("' and TABLE_NAME = '").append(lockTableName).append("'");
        try (Statement statement = connection.createStatement();
             ResultSet resultSet = statement.executeQuery(sql.toString())) {
            if (!resultSet.next()) {
                sql.setLength(0);
                sql.append("CREATE TABLE ").append(lockTableName).append(" (")
                        .append(" `id` bigint(20) auto_increment NOT NULL PRIMARY KEY,")
                        .append(" `lock_time` bigint(20) NOT NULL")
                        .append(") ")
                        .append(" ENGINE=InnoDB ")
                        .append(" DEFAULT CHARSET=utf8mb4 ")
                        .append(" COLLATE=utf8mb4_general_ci;");
                try (Statement lockStatement = connection.createStatement()) {
                    lockStatement.execute(sql.toString());
                }
                return null;
            }
            sql.setLength(0);
            sql.append("select * from ").append(lockTableName);
            try (Statement lockStatement = connection.createStatement();
                 ResultSet lockResultSet = lockStatement.executeQuery(sql.toString())) {
                if (lockResultSet.next()) {
                    long id = lockResultSet.getLong("id");
                    long lockTime = lockResultSet.getLong("lock_time");
                    return new ArchiveLock(id, lockTime);
                }
            }
        }
        return null;
    }

    private void process(ArchiveProperties archiveProperties, Long archiveTimeStamp, Map<String, ArchiveTable> archiveTableMap) {
        if (!archiveProperties.isEnabled()) {
            log.info("archive is not enabled, check the archive switch or the tables configured exists or not");
            return;
        }

        log.info("start process archive tables");
        ArchiveProcessor archiveProcessor = new ArchiveProcessor(archiveProperties);
        // 提取第一级主表，归档时按主表归档
        List<ArchiveTable> archiveTables = new ArrayList<>();
        for (ArchiveTable archiveTable : archiveProperties.getArchiveTables()) {
            String tableName = archiveTable.getTableName();
            if (archiveTableMap.containsKey(tableName)) {
                archiveTables.add(archiveTableMap.get(tableName));
            }
        }
        archiveProcessor.setArchiveTables(archiveTables);
        archiveProcessor.setDataSource(dataSource);
        archiveProcessor.setTargetDataSource(targetDataSource);
        archiveProcessor.setArchiveTableMap(archiveTableMap);
        if (originConnectionProvider != null) {
            archiveProcessor.setOriginConnectionProvider(originConnectionProvider);
        }
        archiveProcessor.process(archiveTimeStamp);
    }

    /**
     * 检查配置信息中的归档表和数据库表是否匹配
     */
    private void checkTableInfo(Map<String, ArchiveTable> propertiesArchiveTables, Map<String, ArchiveTable> archiveTableMap, boolean checkTable) {
        Set<String> propertiesSet = propertiesArchiveTables.keySet();
        Set<String> processSet = archiveTableMap.keySet();
        propertiesSet.forEach(tableName -> {
            if (!processSet.contains(tableName)) {
                Assert.isTrue(!checkTable, () -> "Table " + tableName + " not found in database");
                log.warn("Table {} not found in database", tableName);
            }
        });
    }

    private void extractArchiveTable(List<ArchiveTable> archiveTables, Map<String, ArchiveTable> propertiesArchiveTables) {
        if (archiveTables == null || archiveTables.isEmpty()) {
            return;
        }
        archiveTables.stream().filter(ArchiveTable::isEnabled).forEach(archiveTable -> {
            propertiesArchiveTables.put(archiveTable.getTableName(), archiveTable);
            List<ArchiveTable> relatedTables = archiveTable.getRelatedTables();
            extractArchiveTable(relatedTables, propertiesArchiveTables);
        });
    }

    private void initTableInfo(Set<String> tableNames, Map<String, ArchiveTable> archiveTableMap) throws SQLException {
        Connection connection = dataSource.getConnection();
        PreparedStatement tableStatement = null;
        ResultSet resultSet = null;
        PreparedStatement columnStatement = null;
        try {
            String tableSql = getTableSql(connection.getCatalog(), tableNames);
            tableStatement = connection.prepareStatement(tableSql);
            resultSet = tableStatement.executeQuery();
            columnStatement = connection.prepareStatement("select TABLE_SCHEMA,TABLE_NAME,COLUMN_NAME,COLUMN_COMMENT as REMARKS,COLUMN_KEY " +
                    "from INFORMATION_SCHEMA.COLUMNS where TABLE_SCHEMA = ? and TABLE_NAME = ?");

            while (resultSet.next()) {
                String tableName = resultSet.getString("TABLE_NAME");
                ArchiveTable archiveTable = new ArchiveTable();
                archiveTable.setTableName(tableName);
                archiveTableMap.put(tableName, archiveTable);
                List<ArchiveColumn> archiveColumns = new ArrayList<>();
                columnStatement.setObject(1, connection.getCatalog());
                columnStatement.setObject(2, tableName);
                try (ResultSet columns = columnStatement.executeQuery()) {
                    List<String> columnNames = new ArrayList<>();
                    while (columns.next()) {
                        String columnName = columns.getString("COLUMN_NAME");
                        ArchiveColumn archiveColumn = new ArchiveColumn();
                        archiveColumn.setColumnName(columnName);
                        columnNames.add(columnName);
                        archiveColumns.add(archiveColumn);

                        String columnKey = columns.getString("COLUMN_KEY");
                        if (COLUMN_KEY_PRI.equals(columnKey)) {
                            archiveTable.setPrimaryKey(columnName);
                            archiveTable.setPrimaryKeyIndex(columnNames.size() - 1);
                            archiveColumn.setPrimaryKey(true);
                        }
                    }
                    archiveTable.setColumnNames(columnNames.toArray(new String[0]));
                }
                archiveTable.setColumns(archiveColumns);
                // 表没有主键，自定义列作为主键列处理
                checkCustomPrimaryKeyIndex(archiveTable);
            }
        } finally {
            ArchiveUtil.close(tableStatement, resultSet);
            ArchiveUtil.closeStatement(columnStatement);
            ArchiveUtil.closeConnection(connection);
        }
    }

    private void checkCustomPrimaryKeyIndex(ArchiveTable archiveTable) {
        Optional<ArchiveColumn> optional = archiveTable.getColumns().stream().filter(ArchiveColumn::isPrimaryKey).findFirst();
        if (!optional.isPresent()) {
            String primaryKey = archiveTable.getPrimaryKey();
            Assert.hasText(primaryKey, () -> "Table " + archiveTable.getTableName() + " has no primary key");
            String[] columnNames = archiveTable.getColumnNames();
            for (int i = 0; i < columnNames.length; i++) {
                if (primaryKey.equals(columnNames[i])) {
                    archiveTable.setPrimaryKeyIndex(i);
                    break;
                }
            }
            if (archiveTable.getPrimaryKeyIndex() == null) {
                throw new ArchiveException("Table " + archiveTable.getTableName() + " has no primary key");
            }
        }
    }

    /**
     * 将配置信息保存到归档表
     *
     * @param archiveTables   配置信息
     * @param archiveTableMap 配置提取信息
     */
    private void mergeArchiveTableConfig(List<ArchiveTable> archiveTables, ArchiveTable parent, Map<String, ArchiveTable> archiveTableMap,
                                         Map<String, ArchiveTable> archiveTableWithLevels, boolean useInsertIntoSelect, ArchiveStrategy archiveStrategy) throws IllegalAccessException {
        List<ArchiveTable> relatedTables = new ArrayList<>();
        for (ArchiveTable archiveTable : archiveTables) {
            ArchiveTable dbArchiveTable = archiveTableMap.get(archiveTable.getTableName());
            if (dbArchiveTable != null) {
                String tableName = archiveTable.getTableName();
                String levelTableName = tableName;
                if (parent != null) {
                    levelTableName = parent.getLevelTableName() + "#" + tableName;
                }

                ArchiveTable copyArchiveTable = new ArchiveTable();
                copyProperties(dbArchiveTable, copyArchiveTable);
                copyArchiveTable.setLevelTableName(levelTableName);
                copyArchiveTable.setParentTable(parent);
                mergeTableInfo(copyArchiveTable, archiveTable, useInsertIntoSelect, archiveStrategy);

                List<ArchiveTable> relatedTablesConfig = archiveTable.getRelatedTables();
                if (relatedTablesConfig != null && !relatedTablesConfig.isEmpty()) {
                    mergeArchiveTableConfig(relatedTablesConfig, copyArchiveTable, archiveTableMap, archiveTableWithLevels, useInsertIntoSelect, archiveStrategy);
                }

                relatedTables.add(copyArchiveTable);
                archiveTableWithLevels.put(levelTableName, copyArchiveTable);
            }
        }
        if (parent != null && !relatedTables.isEmpty()) {
            parent.setRelatedTables(relatedTables);
        }
    }

    private void copyProperties(ArchiveTable dbArchiveTable, ArchiveTable copyArchiveTable) throws IllegalAccessException {
        for (Field archiveTableField : archiveTableFields) {
            archiveTableField.set(copyArchiveTable, archiveTableField.get(dbArchiveTable));
        }
    }

    private void mergeTableInfo(ArchiveTable archiveTable, ArchiveTable archiveTableConfig, boolean useInsertIntoSelect, ArchiveStrategy archiveStrategy) {
        archiveTable.setEnabled(archiveTableConfig.isEnabled());
        archiveTable.setShortName(archiveTableConfig.getShortName());
        String primaryKey = archiveTableConfig.getPrimaryKey();
        if (primaryKey != null && !primaryKey.trim().isEmpty()) {
            archiveTable.setPrimaryKey(primaryKey);
        }
        archiveTable.setBatchSize(archiveTableConfig.getBatchSize());
        archiveTable.setArchiveCondition(archiveTableConfig.getArchiveCondition());
        archiveTable.setIndexName(archiveTableConfig.getIndexName());

        extractArchiveCondition(archiveTable, archiveTableConfig);

        // WARNING：分批删除时需要根据主键删除，没有主键需要明确配置primaryKey，删除时按照指定字段删除
        // 表没有主键的情况下需要指定唯一键作为删除字段，否则可能会删错数据
        archiveTable.setExtractColumns(getExtractColumns(archiveTable, archiveTableConfig, useInsertIntoSelect, archiveStrategy));
    }

    private void extractArchiveCondition(ArchiveTable archiveTable, ArchiveTable archiveTableConfig) {
        String archiveCondition = archiveTableConfig.getArchiveCondition();
        Assert.hasText(archiveCondition, () -> "Table " + archiveTable.getTableName() + " archiveCondition cannot null");
        archiveTable.setArchiveCondition(archiveCondition);

        Matcher matcher = ArchiveUtil.getMatcher(archiveCondition);
        while (matcher.find()) {
            String extractCondition = matcher.group(1);
            archiveTable.getExtractConditions().add(extractCondition.trim());
        }
    }


    private List<String> getExtractColumns(ArchiveTable archiveTable, ArchiveTable archiveTableConfig, boolean useInsertIntoSelect, ArchiveStrategy archiveStrategy) {
        String primaryKeyDb = archiveTable.getPrimaryKey();
        List<String> extractColumns = archiveTableConfig.getExtractColumns();
        if (extractColumns == null) {
            extractColumns = new ArrayList<>();
        }
        List<String> columns = new ArrayList<>();
        columns.add(primaryKeyDb);
        extractColumns.forEach(column -> {
            if (!columns.contains(column)) {
                columns.add(column);
            }
        });
        // 使用insert into ... select语法迁移数据时主键放在第0号位置
        if (useInsertIntoSelect && archiveStrategy == ArchiveStrategy.SAME) {
            archiveTable.setPrimaryKeyIndex(0);
        }
        return columns;
    }

    public static String getTableSql(String tableSchema, Set<String> tableNames) {
        StringBuilder tableSql = new StringBuilder("select TABLE_SCHEMA,TABLE_NAME from INFORMATION_SCHEMA.TABLES where TABLE_SCHEMA = '");
        tableSql.append(tableSchema).append("' and TABLE_NAME in (");
        for (String tableName : tableNames) {
            tableSql.append("'").append(tableName).append("',");
        }
        tableSql.deleteCharAt(tableSql.length() - 1);
        tableSql.append(")");
        return tableSql.toString();
    }

}